/**
 * 
 */
package skewreduce.framework;

import java.io.BufferedReader;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.PrintStream;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Writable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import skewreduce.framework.logical.LogicalOp;
import skewreduce.framework.physical.PhysicalOp;
import skewreduce.framework.physical.PhysicalOp.ID;
import skewreduce.framework.planner.PriorityScheduler;
import skewreduce.lib.KahanSum;
import skewreduce.lib.Partition;


public class OperatorSchedule implements Comparator<PhysicalOp>, Writable {
	private static Logger LOG = LoggerFactory.getLogger("PLAN");
	
	ConcurrentHashMap<PhysicalOp.ID,Schedule> manualSchedule = new ConcurrentHashMap<PhysicalOp.ID,Schedule>();
	
	OperatorSchedule() {}
	
	OperatorSchedule(Configuration conf) throws IOException {
		String fn = conf.get("skewreduce.scheduler.manual.file");
		if ( fn != null ) {
			FileReader fr = null;
			BufferedReader reader = null;
			try {
				fr = new FileReader(fn);
				reader = new BufferedReader(fr);
				String line;
				while ( (line = reader.readLine() ) != null ) {
					Schedule s = new Schedule(line.trim());
					manualSchedule.put(s.getID(), s);
				}
			} catch ( FileNotFoundException ignore ) {
				LOG.debug("Schedule file {} is not found. Use default schedule scheme");
			} finally {
				if ( reader != null ) try { reader.close(); } catch ( IOException ignore ) {}
				reader = null;
				if ( fr != null ) try { fr.close(); } catch ( IOException ignore ) {}
				fr = null;
			}
		}
	}
	
	public void addAll(Map<PhysicalOp.ID,Schedule> m) {
		manualSchedule.putAll(m);
	}
	public void add(Schedule s) {
		manualSchedule.put(s.getID(),s);
	}
	public void add(Partition.ID id,double t,double c) {
		PhysicalOp.ID wpid = new PhysicalOp.ID(LogicalOp.Type.WORK,id);
		manualSchedule.put(wpid, new Schedule(wpid,t,c));
	}
	
	public Schedule get(PhysicalOp.ID pid) { return manualSchedule.get(pid); }
	public Schedule get(Partition.ID pid) {
		PhysicalOp.ID wpid = new PhysicalOp.ID(LogicalOp.Type.WORK,pid);
		return manualSchedule.get(wpid);
	}
	public boolean isEmpty() { return manualSchedule.isEmpty(); }
	
	public void clear() {
		manualSchedule.clear();
	}
	
	public PhysicalOp[] sort(Collection<PhysicalOp> ops) {
		PhysicalOp[] arr = new PhysicalOp[ops.size()];
		ops.toArray(arr);
		Arrays.sort(arr,this);
		return arr;
	}

	@Override
	public int compare(PhysicalOp o1, PhysicalOp o2) {
		if ( manualSchedule.isEmpty() ) 
			return o1.compareTo(o2);
		
		Schedule s1 = manualSchedule.get(o1.getID());
		Schedule s2 = manualSchedule.get(o2.getID());
		if ( s1 != null && s2 != null ) {
			return s1.compareTo(s2);
		} else if ( s1 == null && s2 == null ) {
			return o1.compareTo(o2);
		} else {
			PhysicalOp.ID id1 = o1.getID();
			PhysicalOp.ID id2 = o2.getID();
			int rc = id1.compareTo(id2);
			if ( rc != 0 ) return rc;
			// should schedule the one with schedule first
			if ( s1 == null ) {
				return 1; // s2 has schedule
			} else {
				return -1; // s1 has schedule
			}
		}
	}
	
	public void dumpSchedule(String fn) throws IOException {
		if ( manualSchedule.isEmpty() ) return;
		
		Object[] objs = manualSchedule.values().toArray();
		Arrays.sort(objs);
		PrintStream out = new PrintStream(fn);
		for ( Object o : objs ) {
			out.println(o);
		}
		out.close();
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		int n = in.readInt();
        for ( int i = 0; i < n; ++i ) {
        	Schedule s = Schedule.createSchedule(in);
        	add(s);
        }	
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeInt(manualSchedule.size());
		for ( Schedule s : manualSchedule.values() ) {
			s.write(out);
		}
	}
	
	final class PseudoTask implements Comparable<PseudoTask> {
		final Schedule task;
		final double end;
		PseudoTask(Schedule s,double b) {
			task = s;
			end = b + task.getExpectedRuntime();
		}
		
		@Override
		public int compareTo(PseudoTask o) {
			return (int)Math.signum(end - o.end);
		}
	}
	
	public double estimateWorkPhase(int slots) {
		Schedule[] tmp = new Schedule[manualSchedule.size()];
		manualSchedule.values().toArray(tmp);
		Arrays.sort(tmp);
		PriorityQueue<PseudoTask> q = new PriorityQueue<PseudoTask>(slots);
		double t = 0.0;
		int taskIdx = 0;
		while ( ! q.isEmpty() || (taskIdx < tmp.length && tmp[taskIdx].getID().getType() == LogicalOp.Type.WORK) ) {
			while ( q.size() < slots &&  taskIdx < tmp.length && tmp[taskIdx].getID().getType() == LogicalOp.Type.WORK ) {
				q.add(new PseudoTask(tmp[taskIdx],t + tmp[taskIdx].getExpectedRuntime()));
				++taskIdx;
			}
			
			PseudoTask task = q.poll();
			t = task.end;
			while ( ! q.isEmpty() && q.peek().end <= task.end ) {
				q.poll();
			}
		}
		if ( LOG.isDebugEnabled() ) {
			LOG.debug("Estimated work phase = {} seconds",t);
		}
		return t;
	}
	
	public double estimateWorkPhase() {
		KahanSum s = new KahanSum();
		KahanSum s2 = new KahanSum();
		int n = 0;
		for ( Map.Entry<PhysicalOp.ID,Schedule> e : manualSchedule .entrySet() ) {
			if ( e.getKey().getType() == LogicalOp.Type.WORK ) {
				double v = e.getValue().getExpectedRuntime();
				s.add(v);
				s2.add(v*v);
				++n;
			}
		}
		
		double avg = s.value() / n;
		double stdev = Math.sqrt(s2.value() / n  - avg*avg);
		
		if ( LOG.isInfoEnabled() ) {
			LOG.info("WORK phase distribution: avg = {}, stdev = {}",avg,stdev);
		}
		
		return avg + 1.0 * stdev;
	}
}